package com.gmsk.pocket.helper;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.URI;


public class HdfsFileSystem {

    private static final Logger logger = LoggerFactory.getLogger(HdfsFileSystem.class);

    public static void copyFileToHDFSByName(Configuration conf, String hdfsUri, String localFileName, String remoteFileName) throws IOException {
        FileSystem fs = FileSystem.get(URI.create(hdfsUri), conf);
        fs.copyFromLocalFile(new Path(localFileName), new Path(remoteFileName));
        logger.info("copy from local file:{} to HDFS file: {} done.", localFileName, remoteFileName);
        fs.close();
    }

    public static void copyFileToHDFSByFileObj(File localPath, String hdfsPath) throws IOException {
        InputStream in = null;
        if (null == localPath || null == hdfsPath || hdfsPath.isEmpty()) {
            logger.warn("copyFileToHDFSByFile: localpath and hdfspath are required");
            return;
        }
        try {
            Configuration conf = new Configuration();
            FileSystem fileSystem = FileSystem.get(URI.create(hdfsPath), conf);
            FSDataOutputStream out = fileSystem.create(new Path(hdfsPath));
            in = new BufferedInputStream(new FileInputStream(localPath));
            IOUtils.copyBytes(in, out, 4096, false);
            out.hsync();
            out.close();
        }catch (Exception e) {
            e.printStackTrace();
        }finally {
            IOUtils.closeStream(in);
        }
        return;
    }

    public static void downloadFromHDFS(Configuration conf, String uri, String remoteFileName, String localFileName) throws IOException {
        Path path = new Path(remoteFileName);
        FileSystem fs = FileSystem.get(URI.create(uri), conf);
        fs.copyToLocalFile(path, new Path(localFileName));
        fs.close();
        logger.info("downloading file from {} to {} succeed." , remoteFileName, localFileName);
        return;
    }

    public static void downloadFromHDFS(String uri, String HDFSFileName, OutputStream localFileOutPut) throws IOException {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(URI.create(uri), config);
        InputStream in = fs.open(new Path(uri + "/" + HDFSFileName));
        IOUtils.copyBytes(in, localFileOutPut, 4096, true);
        logger.info("downloading HDFS file {} succeed ", HDFSFileName);
        fs.close();
        return;
    }

    public static boolean exists(String HDFSUri, String HDFSFileName) {
        Configuration conf = new HdfsConfiguration();
        boolean fileExists = false;
        try {
            FileSystem fileSystem = FileSystem.get(URI.create(HDFSUri), conf);
            fileExists = fileSystem.exists(new Path(HDFSUri + "/" + HDFSFileName));
        } catch (IOException e) {
            logger.error("hdfs:exist() exception occurs. exception:" + e.getMessage());
            return fileExists;
        }
        logger.info("HDFS URI: {} fileName: {} exists ?" , HDFSUri , HDFSFileName, fileExists);
        return fileExists;
    }
}
